#undef UNICODE
#ifdef _WIN32
#include <ws2tcpip.h>
#else//linux和其它系统
#include <arpa/inet.h>
#include<sys/socket.h>
#include <netinet/in.h>
#include<fcntl.h>
#include<string.h>
#include <unistd.h>
#include<sys/epoll.h>
#define SOCKET int
#endif
#include "server.hpp"
#include <stdio.h>
#include<list>
#include<thread>
#include<mutex>

bool Server::emptysendback(Server*,ThreadInfo*,DataInfo*,void*){return true;}
static unsigned int threadnum=0;
struct Server::ThreadInfo{
	unsigned int id;
	Server* server;
	typedef struct ReceiveRequest{
		DataInfo* info;
		DataPack pack;
	}ReceiveRequest;
	typedef struct SendRequest{
		DataInfo* info;
		DataPack pack;
	}SendRequest;
	std::unordered_set<DataInfo*> sockets,notifysend,notifyrecv,notifyclose;
	std::list<ReceiveRequest> recv;
	std::list<SendRequest> send;
	std::mutex *socket,*pendingrecv,*pendingsend;
	bool exited;
	size_t connected;
	unsigned int sendcount,recvcount,maxpending;
	LostCallback onlost;
	void* lostarg;
	#ifndef _WIN32
	int epollfd;
	#endif

	ThreadInfo():exited(false),sendcount(0),recvcount(0),connected(0),id(threadnum++){}
	~ThreadInfo(){threadnum--;}
};

struct Server::DataInfo{
	struct PackNode{
		DataPack pack;
		PackNode *next;
	};
	PackNode *sendbegin,*sendend,*recvbegin,*recvend;//取begin begin=end end=next
	ThreadInfo* parent;
	size_t sendpos,recvpos;
	time_t recordtime;//等于最大值时自动断开连接
	unsigned int sendcount,recvcount,pentry;
	SOCKET socket;
	char ip[64];
	size_t pos,size;
	char buffer[CACHESIZE];
	//尾部为默认缓冲区
	static DataInfo* create(ThreadInfo* p,SOCKET s,unsigned int entrysize){
		auto v=(DataInfo*)malloc(sizeof(DataInfo)+entrysize);
		v->sendbegin=nullptr,v->sendend=nullptr,
		v->recvbegin=nullptr,v->recvend=nullptr;
		v->parent=p;v->sendpos=0,v->recvpos=0,v->pos=0,v->size=0,v->socket=s;
		v->sendcount=0,v->recvcount=0;
		memset(v->ip,0,64);
		v->recordtime=time(nullptr);
		v->pentry=0;
		return v;
	};
	//调用后析构
	void remove(){
		auto cur=sendbegin,end=sendend;
		while(cur){
			free(cur);
			if(end){
				cur=end;
				end=end->next;
				continue;
			}
			break;
		}
		cur=recvbegin,end=recvend;
		while(cur){
			free(cur);
			if(end){
				cur=end;
				end=end->next;
				continue;
			}
			break;
		}
		#ifdef _WIN32
		closesocket(socket);
		#else
		epoll_event ev;
		epoll_ctl(parent->epollfd,EPOLL_CTL_DEL,socket,&ev);
		close(socket);
		#endif
		free(this);
	}
	void pendsend(DataPack& pack){
		if(sendcount<parent->maxpending){
			sendcount++;
			auto v=(PackNode*)malloc(sizeof(PackNode));
			v->pack=pack;
			v->next=nullptr;
			if(sendend){
				sendend->next=v;
				sendend=v;
				return;
			}
			if(sendbegin){
				sendend=v;
				return;
			}
			sendbegin=v;
		}
	}
	void pendrecv(DataPack& pack){
		if(recvcount<parent->maxpending){
			recvcount++;
			auto v=(PackNode*)malloc(sizeof(PackNode));
			v->pack=pack;
			v->next=nullptr;
			if(recvend){
				recvend->next=v;
				recvend=v;
				return;
			}
			if(recvbegin){
				recvend=v;
				return;
			}
			recvbegin=v;
		}
	}
	size_t getdata(char* recvbuf,size_t len){
		if(len){
			if(pos!=size){
				_thread_begin_get:
				auto got=std::min(len,size-pos);
				memcpy(recvbuf,buffer+pos,got);
				pos+=got;
				return got;
			}
			if(len>=CACHESIZE){
				auto back=recv(socket,recvbuf,len,0);
				if(back>0)return back;
				return -1;
			}
			auto back=recv(socket,buffer,CACHESIZE,0);
			if(back>0){
				pos=0;
				size=back;
				goto _thread_begin_get;
			}
		}
		return 0;
	}
	DataInfo()=delete;
	~DataInfo()=delete;
};

Server::~Server(){}

void Server::sleep(){
	using namespace std;
	this_thread::sleep_for(chrono::milliseconds(1));
}
Server::ThreadInfo* Server::matchleastthread(size_t maxconnect){
	auto threads=this->threads;
	auto l=this->nthread;
	auto ret=&threads[0];
	auto connected=ret->connected;
	for(size_t i=1;i<l;i++){
		auto cur=&threads[i];
		auto v=cur->connected;
		if(v<connected){
			connected=v;
			ret=cur;
		}
	}
	if(connected!=maxconnect)return ret;
	return nullptr;
}
#ifdef _WIN32
bool Server::openlib(){
	WSADATA wsaData;
	int iResult=WSAStartup(MAKEWORD(2,2),&wsaData);
	if(iResult!=0){
		printf("WSAStartup failed with error: %d\n",iResult);
		return false;
	}
	return true;
}
int Server::run(const char* address,unsigned int _port,unsigned int entrysize,DataCallback entryback,
void* entryarg,LostCallback onlost,void* lostarg,unsigned int nthread,
unsigned int maxpending,time_t timeout,size_t maxconnect){
	std::string _p(std::to_string(_port));
	auto port=_p.c_str();
	SOCKET serversocket=INVALID_SOCKET;
	struct addrinfo *result=NULL;
	struct addrinfo hints;
	int iSendResult;
	memset(&hints,0,sizeof(hints));
	hints.ai_family=AF_INET;
	hints.ai_socktype=SOCK_STREAM;
	hints.ai_protocol=IPPROTO_TCP;
	hints.ai_flags=AI_PASSIVE;
	int iResult=getaddrinfo(NULL,port,&hints,&result);
	if(iResult!=0){
		printf("getaddrinfo failed with error: %d\n",iResult);
		WSACleanup();
		return -1;
	}
	serversocket=socket(result->ai_family,result->ai_socktype,result->ai_protocol);
	if(serversocket==INVALID_SOCKET){
		printf("socket failed with error: %d\n",WSAGetLastError());
		freeaddrinfo(result);
		WSACleanup();
		return -1;
	}
	iResult=bind(serversocket,result->ai_addr,(int)result->ai_addrlen);
	if(iResult==SOCKET_ERROR){
		printf("bind failed with error: %d\n",WSAGetLastError());
		freeaddrinfo(result);
		closesocket(serversocket);
		WSACleanup();
		return -1;
	}
	freeaddrinfo(result);
	iResult=listen(serversocket,SOMAXCONN);
	if(iResult==SOCKET_ERROR){
		printf("listen failed with error: %d\n",WSAGetLastError());
		closesocket(serversocket);
		WSACleanup();
		return -1;
	}
	static unsigned long NBMODE=1;
	ioctlsocket(serversocket,FIONBIO,&NBMODE);
	//socket初始化完成
	auto l=nthread==-1?std::thread::hardware_concurrency()-1:nthread-1;
	this->nthread=l;
	threads=new ThreadInfo[l]();
	running=true;
	for(size_t i=0;i<l;i++){
		auto& cur=threads[i];
		cur.server=this,cur.maxpending=maxpending;
		cur.lostarg=lostarg,cur.onlost=onlost;
		cur.pendingrecv=new std::mutex();
		cur.pendingsend=new std::mutex();
		cur.socket=new std::mutex();
		std::thread(_run,this,&cur,entrysize,entryback,entryarg,onlost,lostarg,maxpending,timeout).detach();
	}
	SOCKET clientsocket;
	for(;;){
		_begin_loop:
		update();
		auto cur=matchleastthread(maxconnect);
		if(cur){
			unsigned long ipsize=64;
			sockaddr_storage client={0};
			int addrsize=sizeof(sockaddr_storage);
			clientsocket=accept(serversocket,(struct sockaddr*)&client,&addrsize);
			if(clientsocket!=INVALID_SOCKET){
				ioctlsocket(clientsocket,FIONBIO,&NBMODE);
				auto o=DataInfo::create(cur,clientsocket,entrysize);
				WSAAddressToStringA((struct sockaddr*)&client,addrsize,0,o->ip,&ipsize);
				auto mutex=cur->socket;
				while(!mutex->try_lock())if(!running)goto _end_loop;
				cur->sockets.emplace(o);
				cur->connected++;
				mutex->unlock();
				printf("%s connected\n",o->ip);
				goto _begin_loop;
			}
		}
		if(running)continue;
		_end_loop:
		for(unsigned int i=0;i<l;i++)while(!threads[i].exited)sleep();
		closesocket(serversocket);
		delete[] threads;
		return 0;
	}
}
void Server::closelib(){WSACleanup();}
#else
bool Server::openlib(){return true;}
void Server::closelib(){}
int Server::run(const char* address,unsigned int port,unsigned int entrysize,DataCallback entryback,
void* entryarg,LostCallback onlost,void* lostarg,unsigned int nthread,
unsigned int maxpending,time_t timeout,size_t maxconnect){
	int serversocket=socket(AF_INET,SOCK_STREAM|SOCK_NONBLOCK,IPPROTO_TCP);
	if(serversocket==-1){
		printf("server socket create error:%d\n",errno);
		return -1;
	}
	struct sockaddr_in server_addr{
		.sin_family=AF_INET,
		.sin_port=htons(port)
	};
	server_addr.sin_addr.s_addr=inet_addr(address);
	if(-1==bind(serversocket,(struct sockaddr*)&server_addr,sizeof(sockaddr_in))){
		printf("server socket bind error:%d\n",errno);
		close(serversocket);
		return -1;
	}
	if(listen(serversocket,100)==-1){
		printf("server socket listen error:%d\n",errno);
		close(serversocket);
		return -1;
	}
	auto epollfd=epoll_create(1);
	if(epollfd==-1){
		printf("server epoll create error:%d\n",errno);
		close(serversocket);
		return -1;
	}
	this->epollfd=epollfd;
	epoll_event events[512],ev;
	ev.events=EPOLLET|EPOLLIN;
	if(epoll_ctl(epollfd,EPOLL_CTL_ADD,serversocket,&ev)==-1){
		printf("server epoll ctl error:%d\n",errno);
		close(epollfd);
		close(serversocket);
		return -1;
	}
	struct sockaddr_in client_addr;
	socklen_t client_addr_size=sizeof(sockaddr_in);
	auto l=nthread==-1?std::thread::hardware_concurrency()-1:nthread-1;
	this->nthread=l;
	threads=new ThreadInfo[l]();
	bool erred=false;
	for(size_t i=0;i<l;i++){
		auto& cur=threads[i];
		cur.epollfd=epoll_create(maxconnect);
		if(cur.epollfd==-1)erred=true;
		cur.server=this,cur.maxpending=maxpending;
		cur.lostarg=lostarg,cur.onlost=onlost;
		cur.pendingsend=new std::mutex();
		cur.pendingrecv=new std::mutex();
		cur.socket=new std::mutex();
	}
	if(erred){
		for(size_t i=0;i<l;i++){
			auto& cur=threads[i];
			if(cur.epollfd!=-1)close(cur.epollfd);
			delete cur.pendingsend;
			delete cur.pendingrecv;
			delete cur.socket;
		}
		close(epollfd);
		close(serversocket);
		delete[] threads;
		printf("thread epoll create error:%d\n",errno);
		return -1;
	}
	running=true;
	ev.events=EPOLLIN|EPOLLONESHOT;
	for(size_t i=0;i<l;i++){
		auto& cur=threads[i];
		std::thread(_run,this,&cur,entrysize,entryback,entryarg,
		onlost,lostarg,maxpending,timeout).detach();
	}
	for(;;){
		update();
		auto cur=matchleastthread(maxconnect);
		if(cur){
			auto ret=epoll_wait(epollfd,events,512,10);
			if(ret>=0){
				while(ret--){
					int clientsocket=accept(serversocket,(struct sockaddr*)&client_addr,&client_addr_size);
					if(clientsocket!=-1){
						int flags=fcntl(clientsocket,F_GETFL,0);
						fcntl(clientsocket,F_SETFL,flags|O_NONBLOCK);
						char* ip=inet_ntoa(client_addr.sin_addr);
						auto o=DataInfo::create(cur,clientsocket,entrysize);
						ev.data.ptr=o;
						if(epoll_ctl(cur->epollfd,EPOLL_CTL_ADD,clientsocket,&ev)==-1){
							running=false;
							printf("thread epoll ctl error:%d\n",errno);
							o->remove();
							goto _end_loop;
						}
						auto mutex=cur->socket;
						strcpy(o->ip,ip);
						while(!mutex->try_lock())if(!running){
							o->remove();
							goto _end_loop;
						}
						cur->sockets.emplace(o);
						cur->connected++;
						mutex->unlock();
						printf("%s connected\n",o->ip);
					}
				}
				continue;
			}
			running=false;
			printf("server epoll wait error:%d\n",errno);
		}
		if(running)continue;
		_end_loop:
		for(unsigned int i=0;i<l;i++)while(!threads[i].exited)sleep();
		close(epollfd);
		close(serversocket);
		delete[] threads;
		return 0;
	}
}
#endif

int Server::_run(Server* s,ThreadInfo* info,unsigned int entrysize,
DataCallback entryback,void* entryarg,LostCallback onlost,void* lostarg,
unsigned int maxpending,time_t timeout){
	auto socket=info->socket;
	auto& sockets=info->sockets;
	auto pendingsendlock=info->pendingsend;
	auto pendingrecvlock=info->pendingrecv;
	auto& pendingsend=info->send;
	auto& pendingrecv=info->recv;
	auto& sendcount=info->sendcount;
	auto& recvcount=info->recvcount;
	auto& notifysend=info->notifysend;
	auto& notifyrecv=info->notifyrecv;
	auto& notifyclose=info->notifyclose;
	auto& connected=info->connected;
	#ifndef _WIN32
	auto epollfd=info->epollfd;
	epoll_event events[512],ev;
	ev.events=EPOLLIN|EPOLLONESHOT;
	#endif
	time_t curtime;
	auto begin=notifysend.begin(),end=notifysend.end();
	for(;;){
		while(!pendingsendlock->try_lock())sleep();
		for(auto& v:pendingsend)v.info->pendsend(v.pack);
		pendingsend.clear();
		sendcount=0;
		pendingsendlock->unlock();
		while(!pendingrecvlock->try_lock())sleep();
		for(auto& v:pendingrecv)v.info->pendrecv(v.pack);
		pendingrecv.clear();
		recvcount=0;
		pendingrecvlock->unlock();
		while(!socket->try_lock())sleep();
		for(auto v:notifyclose){
			sockets.erase(v);
			notifyrecv.erase(v);
			notifysend.erase(v);
			onlost(s,info,v,lostarg);
			v->remove();
			connected--;
		}
		notifyclose.clear();
		#ifdef _WIN32
		begin=sockets.begin(),end=sockets.end();
		auto curtime=time(nullptr);
		while(begin!=end){
			auto v=*begin;
			auto p=v->recvbegin;
			if(p){
				if(curtime-v->recordtime<timeout){
					auto precv=&p->pack;
					begin++;
					v->recvpos+=v->getdata(precv->src+v->recvpos,
					precv->size-v->recvpos);
					if(v->recvpos==precv->size){
						v->recvpos=0;
						if(precv->callback(s,info,v,precv->arg)){
							v->recvbegin=v->recvend;
							if(v->recvbegin)v->recvend=v->recvbegin->next;
							v->recvcount--;
							continue;
						}
					}
					continue;
				}
				auto remove=begin;
				begin++;
				sockets.erase(remove);
				onlost(s,info,v,lostarg);
				printf("%s kicked\n",v->ip);
				v->remove();
				connected--;
				continue;
			}
			if(curtime-v->recordtime<timeout){
				begin++;
				v->pentry+=v->getdata((char*)(v+1)+v->pentry,entrysize-v->pentry);
				if(v->pentry==entrysize){
					entryback(s,info,v,entryarg);
					v->pentry=0;
				}
				continue;
			}
			auto remove=begin;
			begin++;
			sockets.erase(remove);
			notifyrecv.erase(v);
			notifysend.erase(v);
			onlost(s,info,v,lostarg);
			printf("%s kicked\n",v->ip);
			v->remove();
			connected--;
		}
		#else
		auto ret=epoll_wait(epollfd,events,512,0);
		if(ret>=0){
			while(ret--){
				auto v=(DataInfo*)events[ret].data.ptr;
				auto got=read(v->socket,v->buffer,CACHESIZE);
				if(got!=-1){
					v->pos=0;
					v->size=got;
					notifyrecv.emplace(v);
					continue;
				}
				sockets.erase(v);
				notifyrecv.erase(v);
				notifysend.erase(v);
				onlost(s,info,v,lostarg);
				v->remove();
				connected--;
			}
		}
		else{
			printf("thread epoll wait error:%d\n",errno);
			s->running=false;
			goto _thread_end_loop;
		}
		begin=notifyrecv.begin(),end=notifyrecv.end();
		while(begin!=end){
			auto v=*begin;
			auto& p=v->recvbegin;
			if(p){
				auto precv=&p->pack;
				auto got=v->getdata(precv->src+v->recvpos,precv->size-v->recvpos);
				if(got!=-1){
					if(got){
						v->recvpos+=got;
						if(v->recvpos==precv->size){
							v->recvpos=0;
							if(precv->callback(s,info,v,precv->arg)){
								v->recvbegin=v->recvend;
								if(v->recvbegin)v->recvend=v->recvbegin->next;
								v->recvcount--;
								continue;
							}
						}
						begin++;
						continue;
					}
					auto remove=begin;
					begin++;
					notifyrecv.erase(remove);
					ev.data.ptr=v;
					epoll_ctl(epollfd,EPOLL_CTL_MOD,v->socket,&ev);
					continue;
				}
				begin++;
				continue;
			}
			auto buffer=(char*)(v+1);
			auto got=v->getdata(buffer+v->pentry,entrysize-v->pentry);
			if(got!=-1){
				if(got){
					v->pentry+=got;
					if(v->pentry==entrysize){
						entryback(s,info,v,entryarg);
						v->pentry=0;
					}
					begin++;
					continue;
				}
				auto remove=begin;
				begin++;
				notifyrecv.erase(remove);
				ev.data.ptr=v;
				epoll_ctl(epollfd,EPOLL_CTL_MOD,v->socket,&ev);
				continue;
			}
			begin++;
		}
		#endif
		begin=notifysend.begin(),end=notifysend.end();
		//发送数据
		while(begin!=end){
			auto v=*begin;
			auto p=v->sendbegin;
			if(p){
				auto psend=&p->pack;
				auto back=send(v->socket,psend->src+v->sendpos,
				psend->size-v->sendpos,0);
				if(back>0){
					v->sendpos+=back;
					if(v->sendpos==psend->size){
						v->sendpos=0;
						if(psend->callback(s,info,v,psend->arg)){
							v->sendbegin=v->sendend;
							if(v->sendbegin)v->sendend=v->sendbegin->next;
							v->sendcount--;
						}
					}
				}
				begin++;
				continue;
			}
			auto remove=begin;
			begin++;
			notifysend.erase(remove);
		}
		#ifndef _WIN32
		begin=sockets.begin(),end=sockets.end();
		curtime=time(nullptr);
		while(begin!=end){
			auto v=*begin;
			if(curtime-v->recordtime>=timeout){
				auto remove=begin;
				begin++;
				sockets.erase(remove);
				notifyrecv.erase(v);
				notifysend.erase(v);
				onlost(s,info,v,lostarg);
				#ifndef _WIN32
					epoll_ctl(epollfd,EPOLL_CTL_DEL,v->socket,events);
				#endif
				connected--;
				printf("%s kicked\n",v->ip);
				v->remove();
				continue;
			}
			begin++;
		}
		#endif
		socket->unlock();
		if(s->running)continue;
		while(!socket->try_lock())sleep();
		_thread_end_loop:
		#ifndef _WIN32
		close(epollfd);
		#endif
		for(auto v:sockets){
			onlost(s,info,v,lostarg);
			v->remove();
		}
		socket->unlock();
		delete info->pendingsend;
		delete info->pendingrecv;
		delete info->socket;
		info->exited=true;
		return 0;
	}
}
void Server::closeclient(ThreadInfo* belong,DataInfo* s){
	auto parent=s->parent;
	if(belong==parent){
		parent->notifyclose.emplace(s);
		return;
	}
	auto m=parent->socket;
	while(!m->try_lock())sleep();
	parent->notifyclose.emplace(s);
	m->unlock();
}
void Server::senddata(ThreadInfo* belong,DataInfo* s,DataPack p){
	auto parent=s->parent;
	if(parent==belong){
		auto count=parent->sendcount;
		if(!count){
			auto back=send(s->socket,p.src,p.size,0);
			if(back==p.size){
				if(p.callback(belong->server,belong,s,p.arg))return;
			}
			else if(back!=-1)s->sendpos=back;
		}
		else if(count==belong->maxpending)return;
		parent->send.push_back({s,p});
		parent->sendcount++;
		return;
	}
	auto m=parent->pendingsend;
	while(!m->try_lock())sleep();
	auto count=parent->sendcount;
	if(!count){
		auto back=send(s->socket,p.src,p.size,0);
		if(back==p.size){
			if(p.callback(parent->server,parent,s,p.arg)){
				m->unlock();
				return;
			}
		}
		else if(back!=-1)s->sendpos=back;
	}
	else if(count==parent->maxpending){
		m->unlock();
		return;
	}
	parent->send.push_back({s,p});
	parent->sendcount++;
	m->unlock();
}
unsigned int Server::threadid(ThreadInfo* from){return from->id;}
void Server::recvdata(ThreadInfo* belong,DataInfo* s,DataPack p){
	auto parent=s->parent;
	if(parent==belong){
		auto count=parent->recvcount;
		if(!count){
			auto back=s->getdata(p.src,p.size);
			if(back==p.size){
				if(p.callback(belong->server,belong,s,p.arg))return;
			}
			else if(back!=-1)s->recvpos=back;
		}
		else if(count==belong->maxpending)return;
		parent->recv.push_back({s,p});
		parent->recvcount++;
		return;
	}
	auto m=parent->pendingrecv;
	while(!m->try_lock())sleep();
	auto count=parent->recvcount;
	if(!count){
		auto back=s->getdata(p.src,p.size);
		if(back==p.size){
			if(p.callback(parent->server,parent,s,p.arg)){
				m->unlock();
				return;
			}
		}
		else if(back!=-1)s->recvpos=back;
	}
	else if(count==parent->maxpending){
		m->unlock();
		return;
	}
	parent->recv.push_back({s,p});
	parent->recvcount++;
	m->unlock();
}
void Server::closeserver(){running=false;}
void Server::resettime(DataInfo* s){
	s->recordtime=time(nullptr);
}
unsigned int Server::threadcount(){return nthread;}
void Server::broadcast(ThreadInfo* belong,DataPack p,std::unordered_set<DataInfo*>& excepts){
	for(auto begin=threads,end=threads+nthread;begin!=end;begin++){
		for(auto v:begin->sockets){
			if(excepts.find(v)==excepts.end()){
				senddata(begin,v,p);
			}
		}
	}
}
const char* Server::getip(DataInfo* info){
	return info->ip;
}

void Server::update(){}
char* Server::getbuffer(DataInfo* info){return (char*)(info+1);}